package com.kingsoft.dc.khaos.module.spark.source

import com.kingsoft.dc.khaos.innertype.Schema
import com.kingsoft.dc.khaos.{DataSetChannel, KhaosConf, KhaosContext, SchemaChannel}
import com.kingsoft.dc.khaos.metadata.{Dependency, KhaosStructField}
import com.kingsoft.dc.khaos.module.SparkModule
import com.kingsoft.dc.khaos.module.spark.metadata.source.{ExtractInfo, ExtractStrategy}
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.DataFrame
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods.{compact, parse, render}

class Source(_mid: String,
             kc: KhaosContext) extends SparkModule with Logging{
  //组件之间的依赖
  override var dependencies: Seq[Dependency] = Seq[Dependency]()
  override var targets:Seq[Dependency] = Seq[Dependency]()

  //获取组件类
  private var strategy: ExtractStrategy = _
  //策略类的接口
  private var sourceStrategy: SourceStrategy = _
  private var module_id: String = _
  private var config: String = _

  private var dbName:String= _
  private var tableName:String = _

  /**
    * 节点逻辑处理的准备信息
    *
    * @param dittoConf : 组件配置信息
    * @param dc        : 组件上下文环境
    */
  override def prepare(dittoConf: KhaosConf,
                       dc: KhaosContext): this.type = {
    log.info("Input_context is :" + dc.toString)
    log.info("Input_mid is :" + _mid)

    //根据_mid获取关于当前组件的配置信息
    val logicConf = getModuleConfiguration() match {
      case Some(conf) => conf
      case None => throw new IllegalAccessException("conf is null")
    }
    implicit val formats = DefaultFormats
    val sourceInfo = parse(logicConf, true).extract[ExtractInfo]
    strategy = sourceInfo.strategy
    module_id = sourceInfo.id
    val strategy_clazz = strategy.clazz

    config = compact(render(strategy.config))

    val configJ = parse(config, true)
    try {
      dbName = (configJ \ "db_name").extract[String]
    } catch {
      case ex: Exception => (dbName = "db_" + scala.util.Random.nextInt(100).toString)
    }
    try {
      tableName = (configJ \ "table_name").extract[String]
    } catch {
      case ex: Exception => (tableName = "table_" + scala.util.Random.nextInt(100).toString)
    }

    try {
      sourceStrategy = Class.forName(strategy_clazz, true, Thread.currentThread().getContextClassLoader)
        .newInstance().asInstanceOf[SourceStrategy]
    } catch {
      case e: Exception => println(e.getStackTrace)
    }
    this
  }

  /**
    * 节点执行逻辑
    *
    * @param dataCollector :  数据采集器
    */
  override def execute(dataCollector: DataSetChannel): this.type = {

    val res = dependencies.isEmpty match {
      case true => sourceStrategy.source(kc, module_id, config, null)
      case false => sourceStrategy.source(kc, module_id, config,  dependencies.head)
    }
    //出度的依赖
    targets.foreach{ target =>
      dataCollector.emit[DataFrame](target, res.asInstanceOf[DataFrame])
    }
    dataCollector.emit[DataFrame](Dependency(null,null,null,null,dbName+"."+tableName), res.asInstanceOf[DataFrame])
    this
  }

  /**
    * 声明输出schema
    *
    * @param declarer :  输出声明器
    */
  override def declareOutputSchema(declarer: SchemaChannel): this.type = {
    val res = dependencies.isEmpty match {
      case true => sourceStrategy.schema(kc, config, null)
      case false => sourceStrategy.schema(kc, config, dependencies.head)
    }
    declarer.declare(Dependency(_mid), new Schema(res.asInstanceOf[List[KhaosStructField]]))
    declarer.declare(Dependency(null,null,null,null,dbName+"."+tableName), new Schema(res.asInstanceOf[List[KhaosStructField]]))
    this
  }

  /**
    * 获取配置信息
    *
    * @return :  返回配置信息
    */
  override def getModuleConfiguration(): Option[String] = {
    kc.conf.getOption(_mid)
  }


  /**
    * 添加trans的信赖
    *
    * @param dependency : 依赖信息，每个组件的唯一组件
    */
  override def addDependencies(dependency: Dependency): this.type = {
    dependencies = dependencies :+ dependency
    this
  }

  def addTarget(dependency: Dependency):this.type ={
    targets = targets :+ dependency
    this
  }
}
